package com.buhler.iot;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.setting.dialect.Props;
import com.buhler.iot.entity.ScriptEntity;
import com.buhler.iot.task.WriteTask;
import com.google.common.collect.Queues;
import com.xxdb.DBConnection;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;

public class MultiConnectionWriteTest {
    static Logger logger = LoggerFactory.getLogger(MultiConnectionWriteTest.class);

    public static void main(String[] args) {
        Props prop = new Props("config.txt");

        int[] rounds = {1, 5, 10, 20, 50, 100, 1000};

        logger.info("preparing 21 DBConnection.....");
        DBConnection[] connections = prepareConnections(prop);

        try{
            for (int size : rounds) {
                executeBatchInsert(size, connections);
            }
        }catch (Exception e){

        }finally {
            closeConnections(connections);
        }

    }

    private static void closeConnections(DBConnection[] connections) {
        for (int i = 0; i < 21; i++) {
            DBConnection connection = connections[i];
            if(connection!=null){
                connection.close();
            }
        }
        logger.info("all collection closed");
    }

    private static DBConnection[] prepareConnections(Props prop) {
        DBConnection[]  connections = new DBConnection[21];
        String host = prop.getStr("host");
        Integer port = prop.getInt("port");
        String userId = prop.getStr("userId");
        String password = prop.getStr("password");
        logger.info("connecting to dolphindb {}:{} with account:{},password:{}", host, port, userId, password);
        for (int i = 0; i < 21; i++) {
            DBConnection connection = new DBConnection();

            try {
                connection.connect(host, port, userId, password);
            }  catch (Exception exception) {
                exception.printStackTrace();
                logger.error(exception.getMessage());
            }
            connections[i]= connection;
        }
        logger.info(" 21 DBConnection ready");
        return connections;
    }


    public static void executeBatchInsert(int size,DBConnection[]  connections ) {


        try {
            logger.info("preparing data ");
            ScriptGenerator generator = new ScriptGenerator(size);
            Set<ScriptEntity> sets = generator.generate();


            logger.info("prepared data with {} rows data each table", size);



            LocalDateTime startTime = LocalDateTime.now();
            logger.info("begin batch insert data");

            for (int round = 0; round < 50; round++) {
                batchInsert(connections,Queues.newConcurrentLinkedQueue(sets));
            }

            LocalDateTime endTime = LocalDateTime.now();
            java.time.Duration duration = java.time.Duration.between(startTime, endTime);
            logger.info("end of batch insert data for 50 rounds.\n every table have {} rows data in each round, total elapse {} ms", size, duration.toMillis());

        } catch (Exception exception) {
            exception.printStackTrace();
            logger.error(exception.getMessage());
        } finally {

        }
    }


    private static void batchInsert(DBConnection[]  connections , ConcurrentLinkedQueue<ScriptEntity> scriptQueue) throws InterruptedException, ExecutionException, TimeoutException {
        ThreadPoolExecutor threadPoolExecutor = ThreadUtil.newExecutor(21, 21);
        List<Future<?>> rets = new ArrayList<>();
        for (int i = 0; i < 21; i++) {
            rets.add(threadPoolExecutor.submit(new WriteTask(connections[i], scriptQueue)));
        }

        for (Future<?> submit : rets) {
            submit.get(60, TimeUnit.SECONDS);
        }
        threadPoolExecutor.shutdown();
    }
}
